package org.dh2580.delay.queue.bean.task;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

import org.dh2580.delay.queue.bean.AbstractMsgListener;
import org.dh2580.delay.queue.bean.db.DBUtil;
import org.dh2580.delay.queue.bean.executor.SimpleExecutor;
import org.dh2580.delay.queue.bean.model.Ack;
import org.dh2580.delay.queue.bean.model.DelayMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author: hao.deng
 * @Date: 2019/5/15-下午1:59
 */
public abstract class SelectMsgTask implements Runnable {

    private static Logger LOG = LoggerFactory.getLogger(SelectExpiredMsgTask.class);

    private SimpleExecutor listenerExecutor;

    private SimpleExecutor callbackExecutor;

    private Map<String, List<AbstractMsgListener>> listeners;

    private Integer maxExpiredMinute;

    private static final String SINGLE_DISTRIBUTE_SQL_TPL = "UPDATE tb_delay_msg_queue SET ack_state = 1 WHERE id = ? AND ack_state = 0;";

    private static final String SINGLE_INIT_SQL_TPL = "UPDATE tb_delay_msg_queue SET ack_state = 0 WHERE id = ?";

    public abstract List<DelayMsg> selectDelayMsgs() throws Exception;

    public SelectMsgTask(SimpleExecutor listenerExecutor, SimpleExecutor callbackExecutor, Map<String, List<AbstractMsgListener>> listeners, Integer maxExpiredMinute) {
        this.listenerExecutor = listenerExecutor;
        this.callbackExecutor = callbackExecutor;
        this.listeners = listeners;
        this.maxExpiredMinute = maxExpiredMinute;
    }

    @Override
    public void run() {
        try {

            List<DelayMsg> msgs = selectDelayMsgs();

            if (msgs == null || msgs.size() == 0) {
                return;
            }

            List<DelayMsg> distributedMsgs = new ArrayList<>(msgs.size());
            for (int i = 0; i < msgs.size(); i++) {
                DelayMsg msg = msgs.get(i);

                int affectRows = DBUtil.execute(SINGLE_DISTRIBUTE_SQL_TPL, msg.getId());
                if (affectRows > 0) {
                    distributedMsgs.add(msg);
                }
            }

            for (int i = 0; i < distributedMsgs.size(); i++) {
                DelayMsg msg = msgs.get(i);

                List<AbstractMsgListener> routeListeners = listeners.get(msg.getRouteKey());
                if (routeListeners == null) {
                    DBUtil.execute(SINGLE_INIT_SQL_TPL, msg.getId());
                    continue;
                }

                AbstractMsgListener routeListener = routeListeners.get(0);

                ListenerTask task = new ListenerTask(routeListener, msg.getMsg());

                Future<Ack> ackFuture = listenerExecutor.submit(task);

                AckCallableTask ackCallableTask = new AckCallableTask(msg.getId(), ackFuture, maxExpiredMinute);
                callbackExecutor.execute(ackCallableTask);
            }
        } catch (SQLException e) {
            LOG.error("Execute Sql Except", e);
        } catch (Exception e) {
            LOG.error("Execute Except", e);
        }
    }

    public Integer getMaxExpiredMinute() {
        return maxExpiredMinute;
    }
}